#pragma once
#include "muduo/proto/dispatcher.h"
#include "muduo/proto/codec.h"
#include "muduo/base/Logging.h"
#include "muduo/base/Mutex.h"
#include "muduo/net/EventLoop.h"
#include "muduo/net/TcpClient.h"
#include "muduo/net/EventLoopThread.h"
#include "muduo/base/CountDownLatch.h"
#include "mq_worker.hpp"
#include "mq_channel.hpp"
namespace mymq
{

    class Connection
    {
    public:
        using Ptr = std::shared_ptr<Connection>;
        typedef std::shared_ptr<google::protobuf::Message> MessagePtr;
        using basicConsumeResponsePtr = std::shared_ptr<basicConsumeResponse>;
        using basicCommonResponsePtr = std::shared_ptr<basicCommonResponse>;
        Connection(const std::string &ip, int port, const AsyncWorker::Ptr &worker)
            : _latch(1), _client(worker->loopthread.startLoop(), muduo::net::InetAddress(ip, port), "Client"),
              _dispatcher(std::bind(&Connection::onUnknownMessage, this, std::placeholders::_1,
                                    std::placeholders::_2, std::placeholders::_3)),
              _codec(std::make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher,
                                                               std::placeholders::_1, std::placeholders::_2, std::placeholders::_3))),
              _worker(worker),
              _channels(std::make_shared<ChannelManager>())
        {
            // 注册消息处理回调
            _dispatcher.registerMessageCallback<basicConsumeResponse>(std::bind(&Connection::BasicConsume, this,
                                                                                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<basicCommonResponse>(std::bind(&Connection::BasicCommon, this,
                                                                               std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            // 设置连接处理回调
            _client.setConnectionCallback(std::bind(&Connection::onConnection, this, std::placeholders::_1));
            // 设置消息回调链
            _client.setMessageCallback(std::bind(&ProtobufCodec::onMessage, _codec.get(),
                                                 std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _client.connect(); // 建立连接
            _latch.wait();
        }
        //创建一个信道并打开信道
        Channel::Ptr OpenChannel()
        {
            auto channel = _channels->CreateChannel(_conn,_codec);
            bool ret = channel->OpenChannel();
            if(ret == false)
            {
                DLOG("打开信道失败");
                return Channel::Ptr();
            }
            return channel;
        }
        void CloseChannel(const Channel::Ptr &channel)
        {
            channel->CloseChannel();
            _channels->DeleteChannel(channel->Cid());
        }

    private:
        void onUnknownMessage(const muduo::net::TcpConnectionPtr &conn, const MessagePtr &message, muduo::Timestamp)
        {
            LOG_INFO << "onUnknownMessage: " << message->GetTypeName();
            conn->shutdown();
        }
        void onConnection(const muduo::net::TcpConnectionPtr &conn)
        {
            if (conn->connected())
            {
                LOG_INFO << "connect success";
                _conn = conn;
                _latch.countDown(); // 关闭阻塞
            }
            else
            {
                LOG_INFO << "connect close";
                _conn.reset(); // 连接关闭
            }
        }
        //将收到的消息进行处理
        void BasicConsume(const muduo::net::TcpConnectionPtr &conn, const basicConsumeResponsePtr &message, muduo::Timestamp)
        {
            //先判断是否存在该信道
            auto channel = _channels->GetChannel(message->cid());
            if(channel.get()==nullptr)
            {
                DLOG("没有找到该信道");
                return;
            }
            _worker->pool.push([channel,message](){
                channel->Consume(message);//由外部用户来决定怎么处理这个消息
            });
        }
        void BasicCommon(const muduo::net::TcpConnectionPtr &conn, const basicCommonResponsePtr &message, muduo::Timestamp)
        {
            //先判断是否存在该信道
            auto channel = _channels->GetChannel(message->cid());
            if(channel.get()==nullptr)
            {
                DLOG("没有找到该信道");
                return;
            }
            //将收到的应答添加进应答处理容器中
            channel->PutBasicResponse(message);
        }

    private:
        muduo::CountDownLatch _latch;            // 实现同步的
        muduo::net::EventLoopThread _loopthread; // 事件监控
        muduo::net::TcpConnectionPtr _conn;      // 客户端对应的连接
        muduo::net::TcpClient _client;           // 客户端
        ProtobufDispatcher _dispatcher;          // 请求分发器对象--向其注册请求处理函数
        ProtobufCodecPtr _codec;                 // 协议处理器，针对不同的请求使用不同的处理方法
        ChannelManager::Ptr _channels;
        AsyncWorker::Ptr _worker;
    };
}